import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
//注意,这种方式是基于反射的方式
// 运行前修改配置
// [[ -n "$JAVA_OPTS" ]] || JAVA_OPTS="-Xmx1024M -Xms256M"
//运行方法:
// scalac RDD2Dataframe.scala
// scala RDD2DataFrameByReflection
// rm *.class


object RDD2DataFrameByReflection 
{
  //case class 要放在main方法外面
  case class Person(name: String, salary: Int)

  
  def main(args: Array[String]): Unit = 
  {
    val conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameByReflection")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    // val sqlContext=new  SparkSession.builder(sc)# 如果使用这种最新的用法,会与后面的toDF()函数冲突,暂时不知道如何正确使用



    // 导入语句，可以隐式地将RDD转化成DataFrame

    import sqlContext.implicits._
    val lines = sc.textFile("file:////home/appleyuchi/Desktop/码云/spark_success/Dataframe与RDD互相转化/employee.txt")
    val df = lines.map(_.split(",")).map(x=>Person(x(0),x(1).trim().toInt)).toDF()//toDF的意思是转化为Dataframe
    // df.registerTempTable("employee")用下面一句代码代替,否则会有警告
    df.createOrReplaceTempView("employee")
    val bigDatas = sqlContext.sql("select * from employee where salary >= 10000")
    val personList = bigDatas.javaRDD.collect()
    for (p <- personList.toArray) //这里<-是遍历循环的意思
    {
      println(p)
    }
  }
}


// 参考文献:
// https://blog.csdn.net/m0_37996943/article/details/78553033
//https://blog.csdn.net/sundujing/article/details/51438142

//关于反射的方式的说法,下面链接指出了不足
//https://blog.csdn.net/lemonzhaotao/article/details/80296314
